package com.xuelang.pipeline.app.log.client;

import lombok.extern.slf4j.Slf4j;
import org.java_websocket.client.WebSocketClient;
import org.java_websocket.handshake.ServerHandshake;
import org.springframework.web.socket.TextMessage;
import org.springframework.web.socket.WebSocketSession;

import java.net.URI;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

/**
 * <p>日志webSocket客户端</p>
 *
 * @author : yixin.lys
 * @date : 2024-07-01 09:59
 **/
@Slf4j
public class LogWebSocketClient extends WebSocketClient {

    private WebSocketSession session;
    private ScheduledExecutorService heartbeatExecutor;

    public LogWebSocketClient(URI serverUri, WebSocketSession session, ScheduledExecutorService heartbeatExecutor) {
        super(serverUri);
        this.session = session;
        this.heartbeatExecutor = heartbeatExecutor;
    }

    @Override
    public void onOpen(ServerHandshake serverHandshake) {
        log.info("Connected to log provider server");

        // 启动心跳机制
        heartbeatExecutor.scheduleAtFixedRate(() -> {
            try {
                if (this.isOpen()) {
                    this.send("ping");
                    log.info("Sent ping to log provider server");
                }
            } catch (Exception e) {
                log.error("Error sending ping: " + e.getMessage(), e);
            }
        }, 0, 30, TimeUnit.SECONDS);
    }

    @Override
    public void onMessage(String message) {
        try {
            if ("pong".equals(message)) {
                log.info("Received pong from log provider server");
            } else {
                session.sendMessage(new TextMessage(message));
            }
        } catch (Exception e) {
            log.error("Error processing message: " + e.getMessage(), e);
        }
    }

    @Override
    public void onClose(int code, String reason, boolean remote) {
        log.info("Disconnected from log provider server");
        shutdownHeartbeatExecutor();
    }

    @Override
    public void onError(Exception e) {
        log.error("WebSocket error: " + e.getMessage(), e);
        shutdownHeartbeatExecutor();
    }

    private void shutdownHeartbeatExecutor() {
        if (heartbeatExecutor != null && !heartbeatExecutor.isShutdown()) {
            heartbeatExecutor.shutdownNow();
        }
    }
}
